package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.utils.HbaseUtil;
import com.atguigu.gmall.realtime.utils.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Felix
 * @date 2023/7/10
 * 互动域：评论事实表
 * 需要启动的进程
 * zk、kafka、maxwell、hdfs、hbase、DwdInteractionCommentInfo
 */
public class DwdInteractionCommentInfo {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2.检查点相关的设置(略)
        //TODO 3.从topic_db主题中读取数据  创建动态表
        String topic = "topic_db";
        String groupId = "dwd_interaction_comment_group";
        tableEnv.executeSql(KafkaUtil.getTopicDbDDL(groupId));
        // tableEnv.executeSql("select * from topic_db").print();

        //TODO 4.过滤出评论表数据
        Table commentInfo = tableEnv.sqlQuery("select\n" +
            "    `data`['id'] id,\n" +
            "    `data`['user_id'] user_id,\n" +
            "    `data`['sku_id'] sku_id,\n" +
            "    `data`['appraise'] appraise,\n" +
            "    `data`['comment_txt'] comment_txt,\n" +
            "    ts,\n" +
            "    proc_time \n" +
            "from\n" +
            "    topic_db\n" +
            "where \n" +
            "    `table`='comment_info' and `type`='insert'");
        tableEnv.createTemporaryView("comment_info",commentInfo);
        // tableEnv.executeSql("select * from comment_info").print();

        //TODO 5.从hbase的字典表中读取数据 创建动态表
        tableEnv.executeSql(HbaseUtil.getBaseDicLookUpDDL());
        // tableEnv.executeSql("select rowkey dic_code,info.dic_name from base_dic").print();
        //TODO 6.将评论和字段进行关联
        Table resTable = tableEnv.sqlQuery("SELECT \n" +
            "    id,\n" +
            "    user_id,\n" +
            "    sku_id,\n" +
            "    appraise,\n" +
            "    dic.dic_name appraise_name,\n" +
            "    comment_txt,\n" +
            "    ts\n" +
            "FROM comment_info ci JOIN base_dic FOR SYSTEM_TIME AS OF ci.proc_time dic\n" +
            "ON ci.appraise = dic.rowkey");
        tableEnv.createTemporaryView("res_table",resTable);
        // tableEnv.executeSql("select * from res_table").print();

        //TODO 7.将关联的结果写到kafka主题
        //7.1 创建一个动态表和要写入的主题进行映射
        tableEnv.executeSql("CREATE TABLE dwd_interaction_comment (\n" +
            "  id string,\n" +
            "  user_id string,\n" +
            "  sku_id string,\n" +
            "  appraise string,\n" +
            "  appraise_name string,\n" +
            "  comment_txt string,\n" +
            "  ts string,\n" +
            "  PRIMARY KEY (id) NOT ENFORCED\n" +
            ") " + KafkaUtil.getUpsertKafkaDDL("dwd_interaction_comment"));
        //7.2 写入
        tableEnv.executeSql("insert into dwd_interaction_comment select * from res_table");

    }
}
